/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.lidejia.hadooptotalsort.meta;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

/**
 *
 * @author lidejia
 */
public class SortKeyMetaIO {

    public static void writeTo(Path path, String metas) {
        try {
            FileSystem fs = FileSystem.get(new Configuration());
            // write to hdfs
            try {
                PrintWriter writer = new PrintWriter(new OutputStreamWriter(fs.create(path)));
                try {
                    writer.println(metas);
                } finally {
                    writer.close();
                }
            } finally {
                fs.close();
            }
            // write to local, used by sampler
            fs = FileSystem.getLocal(new Configuration());
            try {
                PrintWriter writer = new PrintWriter(new OutputStreamWriter(fs.create(new Path("_metas"))));
                try {
                    writer.println(metas);
                } finally {
                    writer.close();
                }
            } finally {
                fs.close();
            }
        } catch (IOException ex) {
            Logger.getLogger(SortKeyMetaIO.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public static List<SortKeyMeta> readFrom(Path path) {
        List<SortKeyMeta> metaList = new ArrayList<SortKeyMeta>();
        try {
            //
            FileSystem fs = FileSystem.getLocal(new Configuration());
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(path)));
                try {
                    String input = reader.readLine();
                    //
                    String[] metas = input.split(";");

                    for (String meta : metas) {
                        String[] tokens = meta.split(":");
                        int index = Integer.parseInt(tokens[0]);
                        Class<? extends WritableComparable> type = null;
                        if (tokens[1].equals("int")) {
                            type = IntWritable.class;
                        } else if (tokens[1].equals("long")) {
                            type = LongWritable.class;
                        } else if (tokens[1].equals("string")) {
                            type = Text.class;
                        }
                        boolean reverse = (tokens[2].equals("asc")) ? false : true;
                        metaList.add(new SortKeyMeta(type, index, reverse));

                    }
                } finally {
                    reader.close();
                }
            } finally {
                fs.close();

            }

        } catch (IOException ex) {
            Logger.getLogger(SortKeyMetaIO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return metaList;
    }
}
